package com.by.mqsender;

import com.by.mqsender.config.ByMqConfig;
import com.by.mqsender.consts.ByMqConst;
import com.by.mqsender.properties.ByMqProperties;
import com.by.mqsender.support.ByMqSupport;
import com.google.common.collect.Sets;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.StaticSessionCredentialsProvider;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;
import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.Map;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Collectors;

import static com.by.toolkit.utils.ExceptionUtil.getStackTrace;

/**
 * mq消息发送者自动装配
 *
 * @author YanBingHao
 * @since 2025/9/29
 */
@Slf4j
@ComponentScan({"com.by.mqsender"})
@Configuration
@ConditionalOnProperty(name = "by.mq.enabled", havingValue = "true")
@EnableConfigurationProperties({ByMqProperties.class})
public class MqSenderAutoConfiguration {

    @Resource
    private ByMqProperties byMqProperties;

    @Resource
    private ByMqConfig byMqConfig;

    @Bean
    public ByMqSupport initProducer() {
        String endpoint = byMqProperties.getEndpoint();
        String instanceName = byMqProperties.getInstanceName();
        String instancePwd = byMqProperties.getInstancePwd();
        Set<String> topicProducers = byMqProperties.getTopicProducers();
        log.info("[MqSenderAutoConfiguration][initProducer] byMqProperties:[{}]", byMqProperties);


        topicProducers = CollectionUtils.isEmpty(topicProducers)
                ? Sets.newHashSet()
                : topicProducers;

        topicProducers.add(ByMqConst.DEFAULT_PRODUCER);
        topicProducers.add(ByMqConst.DEFAULT_DELAY_PRODUCER);

        ClientServiceProvider provider = ClientServiceProvider.loadService();
        Map<String, Producer> producerMap = topicProducers.stream().collect(Collectors.toMap(Function.identity(), i -> {
            try {
                return provider.newProducerBuilder()
                        .setClientConfiguration(ClientConfiguration.newBuilder()
                                .setEndpoints(endpoint)
                                .setCredentialProvider(new StaticSessionCredentialsProvider(instanceName, instancePwd))
                                .setRequestTimeout(Duration.of(3000L, ChronoUnit.MILLIS))
                                .build())
                        .setMaxAttempts(3)
                        .build();
            } catch (Exception e) {
                throw new RuntimeException("创建生产者失败：" + getStackTrace(e));
            }
        }));
        return new ByMqSupport(producerMap, byMqConfig);
    }


}
